package cn.com.sparkle.paxos.userserver;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.ConfigNode;
import cn.com.sparkle.paxos.Configuration;
import cn.com.sparkle.paxos.client.MasterMayBeLostException;
import cn.com.sparkle.paxos.client.PaxosClient;
import cn.com.sparkle.paxos.client.PaxosOperater;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.MasterChangePosEvent;
import cn.com.sparkle.paxos.net.factory.NetCloseException;
import cn.com.sparkle.paxos.net.factory.NetHandler;
import cn.com.sparkle.paxos.net.factory.PaxosSession;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MasterBeatHeartRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;

public class UserServerHandler implements NetHandler {

	private final static Logger logger = Logger
			.getLogger(UserServerHandler.class);
	private volatile boolean isMaster = false;
	private volatile String masterAddress;
	private ConcurrentHashMap<PaxosSession, PaxosSession> wantConnectMasterSessionMap = new ConcurrentHashMap<PaxosSession, PaxosSession>();
	private ReentrantLock masterChangelock = new ReentrantLock();
	private UserHandlerInterface handlerInterface;
	private volatile PaxosClient client = null;

	public UserServerHandler(EventsManager eventsManager,
			final Configuration conf, UserHandlerInterface handlerInterface) {
		this.handlerInterface = handlerInterface;
		eventsManager.registerEvent(new MasterChangePosEvent() {
			@Override
			public void masterChange(String address) {
				// nothing to do
				try {
					masterChangelock.lock();
					logger.debug("master changed");
					for (ConfigNode cnode : conf.getSenators()) {
						if ((cnode.getIp() + ":" + cnode.getPort())
								.equals(address)) {
							masterAddress = cnode.getIp() + ":"
									+ cnode.getClientPort();
						}
					}

					if (client == null) {
						try {
							client = new PaxosClient(
									new String[] { masterAddress },conf.getFilePath() + "/service_out_net.prop",conf.getNetLayer(), conf
											.getTransportTcpNum(), true);
						} catch (Throwable e) {
							logger.error("fatal error", e);
						}
					} else {
						logger.debug("change address");
						try {
							client.changeSenator(new String[] { masterAddress });
						} catch (Throwable e) {
							logger.error("fatal error", e);
						}
					}
				} finally {
					masterChangelock.unlock();
				}
			}

			@Override
			public void lostPos() {
				try {
					masterChangelock.lock();
					isMaster = false;
					for (PaxosSession session : wantConnectMasterSessionMap
							.keySet()) {
						session.closeSession();
					}
				} finally {
					masterChangelock.unlock();
				}
			}

			@Override
			public void getMasterPos() {
				try {
					masterChangelock.lock();
					isMaster = true;
				} finally {
					masterChangelock.unlock();
				}
			}
		});
	}

	@Override
	public void onDisconnect(PaxosSession session) {
		wantConnectMasterSessionMap.remove(session);
		UserPaxosSession uSession = (UserPaxosSession)session.attachment();
		handlerInterface.onClientClose(uSession);
	}

	@Override
	public void onConnect(PaxosSession session) {
		UserPaxosSession uSession = new UserPaxosSession(session);
		session.attach(uSession);
		handlerInterface.onClientConnect(uSession);
	}

	@Override
	public void onRecieve(final PaxosSession session, MessagePackage messagePackage) {
			
			final long packageId = messagePackage.getId();
			if(messagePackage.hasCustomOnlyConnectMaster()){
				try {
					masterChangelock.lock();
					if (isMaster) {
						wantConnectMasterSessionMap.put(session, session);
					} else {
						session.closeSession();
					}
				} finally {
					masterChangelock.unlock();
				}
			}else if (messagePackage.hasAddRequest()) {
				
				AddRequest addRequest = messagePackage.getAddRequest();
				UserPaxosSession uSession = (UserPaxosSession)session.attachment();
				
				if (addRequest.getTransportMaster()
						|| addRequest.getIsManagerCommand()) {
					if (isMaster) {
						if(uSession.getPaxosOperater() != null){
							try {
								uSession.getPaxosOperater().waitAllFinish(0);
								uSession.setPaxosOperater(null);// cancel the channel assigned to this session 
							} catch (InterruptedException e) {
								logger.error("unexcepted error", e);
							} catch (MasterMayBeLostException e) {
								//unreachable
							}
						}
						handlerInterface.onClientCustomCommandReceive(uSession,
								messagePackage);
					} else {
						if (client != null) {
							try {
								PaxosOperater operator = uSession.getPaxosOperater();
 								if (operator == null) {
									operator = client.getOperator();
									uSession.setPaxosOperater(operator);// assign
																		// a
																		// channel
																		// to
																		// master
								}
								try {
									operator.add(addRequest.getValue()
											.toByteArray(), 0, true,
											new PaxosOperater.CallBack() {
												@Override
												public void callBack(
														AddResponse response) {
													try {
														MessagePackage.Builder mp = MessagePackage
																.newBuilder()
																.setIsLast(true)
																.setId(packageId);
														mp.setAddResponse(response);
														session.write(mp
																.build());
													} catch (NetCloseException e) {
													}
												}
											});
								} catch (MasterMayBeLostException e) {
									throw new RuntimeException("the master may be lost!");
								}
							} catch (InterruptedException e) {
								logger.error("fatal error", e);
								;
							}
						} else {
							session.closeSession();
						}
					}
				} else {
					handlerInterface.onClientCustomCommandReceive(uSession,
							messagePackage);
				}
			} else if (messagePackage.hasMasterBeatHeartRequest()) {
				// send master beatheart request
				MessagePackage.Builder mp = MessagePackage
						.newBuilder()
						.setIsLast(true)
						.setId(packageId)
						.setMasterBeatHeartRequest(
								MasterBeatHeartRequest.getDefaultInstance());
				try {
					session.write(mp.build());
				} catch (NetCloseException e) {
					// nothing to do
				}
			} else {
				logger.error("unexcepted message");
			}
	}

	@Override
	public void onRefuse(Object connectAttachment) {
		//this is a server, nothing to do
	}

}
